package com.shujia.flink.state

import org.apache.flink.api.common.functions.{RichMapFunction, RuntimeContext}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup

object Demo3Checkpoint {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * checkpoint ： 定时将flink执行过程中的状态保存到状态后端
      * 只保存最新一次成功的checkpoint
      *
      *
      * 任务执行失败之后可以从最新的checkpoint恢复任务
      * flink run -c com.shujia.flink.state.Demo3Checkpoint -s hdfs://master:9000/flink/checkpoint/a306d86
      * 90ca7718aec6a425b0d8ee171/chk-645  flink-1.0.jar
      *
      * -s： 指定恢复任务的未知，需要到chk-***
      *
      *
      */
    // 每 1000ms 开始一次 checkpoint
    env.enableCheckpointing(1000)

    // 高级选项：
    // 设置模式为精确一次 (这是默认值)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    // 确认 checkpoints 之间的时间会进行 500 ms
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

    // Checkpoint 必须在一分钟内完成，否则就会被抛弃
    env.getCheckpointConfig.setCheckpointTimeout(60000)

    // 同一时间只允许一个 checkpoint 进行
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

    //当作业取消时，保留作业的 checkpoint。注意，这种情况下，需要手动清除该作业保留的 checkpoint。
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

    //将状态保存到hdfs的状态后端
    val stateBackend = new FsStateBackend("hdfs://master:9000/flink/checkpoint")

    //设置状态后端
    env.setStateBackend(stateBackend)


    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val kvDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))

    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    val countDS: DataStream[(String, Int)] = keyByDS.map(new RichMapFunction[(String, Int), (String, Int)] {


      /**
        * ValueState： 单值状态，为每一个key保存一个值
        * 保证在状态中的数据不会因为flink程序挂断而丢失
        * checkpoint不能再本地执行，需要将代码提交到服务器中，状态的数据需要保存再hdfs中
        *
        */
      var valueState: ValueState[Int] = _

      /**
        * open:方法在map之前执行，一般用于建立网络链接
        * 还用于初始化状态
        *
        */
      override def open(parameters: Configuration): Unit = {

        //1、获取flink的环境对象,通过环境初始化状态
        val context: RuntimeContext = getRuntimeContext

        //创建状态的描述对象，需要指定状态保存的数据类型和状体名
        val valueStateDesc = new ValueStateDescriptor[Int]("count", classOf[Int])

        //获取状态
        valueState = context.getState(valueStateDesc)
      }

      override def map(value: (String, Int)): (String, Int) = {

        /**
          * 使用状态统计单词的数量
          *
          */

        //获取之前的单词数量
        val count: Int = valueState.value()

        //计算新的数量
        val newCount: Int = count + 1

        //更新状态
        valueState.update(newCount)

        //返回数据
        (value._1, newCount)
      }
    })

    countDS.print()

    env.execute()


  }

}
